package com.example.homework1

import java.util.concurrent.TimeUnit

import com.example.homework1.MyKafkaConsumer.getKafkaProducerParams
import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}

import scala.collection.mutable
import scala.io.Source

/**
 * @author lbq
 * @date 2022/2/20 20:59
 * @version 1.0
 */
object MyKafkaProducer {
  def main(args: Array[String]): Unit = {

    val source = Source.fromFile("data/sample.log")
    val lines = source.getLines()

    val records: mutable.Buffer[String] = lines.toBuffer

    // 定义 kafka 参数
    val topic = "topic_A1"

    val producer = new KafkaProducer[String, String](getKafkaProducerParams)

    for (i <- records.indices) {
      val record: String = records(i)
      val msg: ProducerRecord[String, String] = new ProducerRecord[String, String](topic, i.toString, record)
      producer.send(msg)
//      TimeUnit.MICROSECONDS.sleep(100)
    }

    producer.close()
    source.close()
  }
}
